
#include "DefaultThreadPool.h"

using namespace ttg::structures;
using namespace ttg_internal::threads;

//-----------------
//--- Performer ---
//-----------------

Task *Performer::getTask_ts()
{
	Task *res = NULL;
	//Trying to get local task (without synchronization).
	if (local_tasks.size() != 0)
	{
		res = local_tasks[local_tasks.size() - 1];
		local_tasks.pop_back();
	}
	//Trying to get root task (with global synchronization).
	else
	{
		tasks_locker.enter();
		if (tasks.size() != 0)
		{
			res = tasks[tasks.size() - 1];
			tasks.pop_back();
		}
		tasks_locker.leave();
	}
	return res;
}

void Performer::work()
{
	this->sleepDeep();
	while (true)
	{
		Task *cur = getTask_ts();
		//If there aren't any task, trying to get new tasks.
		if (cur == NULL)
		{
			//Getting tasks.
			is_gonna_sleep_locker.enter();
			if ((cur = on_idle.notify(this)) == NULL)
				is_gonna_sleep = true;
			is_gonna_sleep_locker.leave();
			//Or sleeping.
			if (cur == NULL)
			{
				bool is_first = true;
				do
				{
					if (is_first)
					{
						//sleepFast(0.25);			//like Chuck Norris - he never sleeps, he waits ...
						is_first = false;
					}
					else
						sleepDeep();
				}
				while ((cur = getTask_ts()) == NULL);
			}
		}
		//Performing task.
		cur->perform(local_tasks);
		SAFE_DELETE(cur);
	}
}

Performer::Performer(size_t core)
	:ttg::threads::Thread(true)
{
	is_gonna_sleep = false;
	this->setTag((int)core);
	setIdealCore(core);
}

void Performer::addTask(Task *task)
{
	//Adding task.
	tasks_locker.enter();
	tasks.push_back(task);
	tasks_locker.leave();
	//Waking.
	if (!is_gonna_sleep_locker.tryEnter())
		return;
	if (is_gonna_sleep || isSleeping())
	{
		is_gonna_sleep = false;
		wake();
	}
	is_gonna_sleep_locker.leave();
}

bool Performer::isIdle()
{
	bool res;
	if (!is_gonna_sleep_locker.tryEnter())
		return false;
	res = is_gonna_sleep || isSleeping();
	is_gonna_sleep_locker.leave();
	return res;
}

Performer::~Performer()
{
	//Releasing root tasks.
	tasks_locker.enter();
	for (size_t i = 0; i < tasks.size(); i++)
		SAFE_DELETE(tasks[i]);
	tasks_locker.leave();
	//Releasing local tasks.
	for (size_t i = 0; i < local_tasks.size(); i++)
		SAFE_DELETE(local_tasks[i]);
}

//-------------------------
//--- DefaultThreadPool ---
//-------------------------

Task *DefaultThreadPool::onIdlePerformer(DefaultThreadPool *owner, Performer *smoker)
{
	if (!owner->giveTaskTo_ts(smoker))
	{
		return owner->onIdle().notify(owner->getIdlePerformerCount());
	}
	return NULL;
}

bool DefaultThreadPool::giveTaskTo_ts(Performer *performer)
{
	bool res = false;
	untaken_tasks_locker.enter();
	if (untaken_tasks.size() != 0)
	{
		performer->addTask(untaken_tasks[untaken_tasks.size() - 1]);
		untaken_tasks.pop_back();
		res = true;
	}
	untaken_tasks_locker.leave();
	return res;
}

void DefaultThreadPool::distributeBetweenSleepers(std::vector<Task *> &tasks)
{
	//Searching for sleeping performer.
	for (size_t i = 0; i < performers.size() && tasks.size() != 0; i++)
		if (performers[i]->isIdle())
		{
			//Given tasks to it.
			performers[i]->addTask(tasks[tasks.size() - 1]);
			tasks.pop_back();
		}
}

DefaultThreadPool::DefaultThreadPool(size_t performer_count)
	:ThreadPool(performer_count)
{
	performers.resize(performer_count);
	for (size_t i = 0; i < performer_count; i++)
	{
		performers[i] = new Performer(i);
		performers[i]->onIdle() += make_delegate(this, onIdlePerformer);
	}
}

size_t DefaultThreadPool::getIdlePerformerCount()
{
	size_t res = 0;
	for (size_t i = 0; i < performers.size(); i++)
		if (performers[i]->isSleeping())
			res++;
	return res;
}

void DefaultThreadPool::addTasks(ttg::structures::Enumerator<Task *> &tasks)
{
	untaken_tasks_locker.enter();
	//Storing.
	while (tasks.moveNext())
	{ untaken_tasks.push_back(tasks.current()); }
	//Getting tasks to sleepers.
	distributeBetweenSleepers(untaken_tasks);
	untaken_tasks_locker.leave();
}

DefaultThreadPool::~DefaultThreadPool()
{
	//Destroyng untaken tasks.
	untaken_tasks_locker.enter();
	for (size_t i = 0; i < untaken_tasks.size(); i++)
		SAFE_DELETE(untaken_tasks[i]);
	untaken_tasks.clear();
	untaken_tasks_locker.leave();
	//Destroying performers.
	for (size_t i = 0; i < performers.size(); i++)
	{
		performers[i]->terminate();
		SAFE_DELETE(performers[i]);
	}
}
